package org.fujene.search;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Logger;

import org.fujene.communicate.local.LocalPeerSystem;
import org.fujene.constants.Constants;
import org.fujene.structure.Enumerates;
import org.fujene.structure.IndexQueue;
import org.fujene.structure.Result;
import org.msgpack.MessagePackObject;
import org.msgpack.object.ArrayType;
import org.msgpack.object.NilType;
import org.msgpack.object.RawType;
import org.msgpack.rpc.Request;
import org.msgpack.rpc.Server;
import org.msgpack.rpc.loop.EventLoop;

/**
 * サーチエンジンです。 クライアントからの検索クエリを受け入れて、本体に投入します。
 * レプリケーション非対応バージョン。
 */
public class SearchSystem{
	// executor専用
	public int EXECUTOR_THREADS = 1;
	// public static ArrayList<String> queueList = null;
	public boolean stopFlag = false;
	public volatile int threadCounter = 0;
	private LocalPeerSystem myPeer = null;
	private ExecutorService exec = null;
	private Logger logger = null;

	public SearchSystem(LocalPeerSystem myPeer, Logger logger){
		// 待ちうけスレッド起動
		exec = Executors.newSingleThreadExecutor();
		exec.execute(new SearchMainService());
		this.myPeer = myPeer;
		this.logger = logger;
	}

	public class SearchRPCHandler{
		public void search(Request req) throws Exception{
			MessagePackObject obj = req.getArguments();
			MessagePackObject[] args = obj.asArray();
			String query = args[0].asString();

			try{
				// Future threadで投げたクエリと戻りの対応をとる
				Result result;
				result = myPeer.setNewSearch(new IndexQueue(Enumerates.SEARCH, query, System
						.currentTimeMillis()));

				MessagePackObject[] objs = new MessagePackObject[result.results.length];
				int count = 0;
				for(String str: result.results){
					objs[count++] = RawType.create(str);
				}

				req.sendResult(ArrayType.create(objs));
			}
			catch(InterruptedException e){
				e.printStackTrace();
			}
		}

		public void stop(Request req){
			this.notifyAll();
			exec.shutdownNow();
			req.sendResult(NilType.create());
		}
	}

	public class SearchMainService implements Runnable{
		public SearchMainService(){}

		@Override
		public void run(){
			logger.fine("SearchMainService started.");
			Server server = new Server(EventLoop.defaultEventLoop());
			try{
				server.serve(new SearchRPCHandler());
				server.listen(Constants.SEARCH_PORT);

				logger.fine("Search server port " + Constants.SEARCH_PORT + " has established.");

				synchronized(this){
					this.wait();
				}

				logger.fine("SearchMainService ended.");

				server.close();
			}
			catch(Exception ignored){}
		}
	}
}